package com.controller.consumer.elasticsearch;

import com.bean.consume.ElasticSearchBean;
import com.dms.conf.DmsConfig;
import com.dms.consumer.DmsConsumer;
import com.service.consume.ElasticSearchService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import java.util.Arrays;

@Controller
@RequestMapping("/es")
public class ElasticSearchController {

    @Autowired
    private ElasticSearchService elasticSearchService;

    @RequestMapping("/consume")
    @ResponseBody
    public String consumerElasticSearch() throws Exception {
        String path = DmsConfig.CONFIG_CONSUMER_FILE_NAME;
        String groupId = "group_Es";
        DmsConsumer consumer = new DmsConsumer(groupId, path);
        consumer.consume(Arrays.asList("topic-Log", "topic-App"));
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < 5; i++) {
            ConsumerRecords<Object, Object> records = consumer.poll(1000);
            System.out.println("the numbers of topic:" + records.count());
            sb.append("the numbers of topic:").append(records.count());
            for (ConsumerRecord<Object, Object> record : records) {
                elasticSearchService.insertElasticSearchBean(new ElasticSearchBean(record));
                System.out.println("consume ElasticSearch:" + record.toString());
                sb.append(record);
            }
        }
        return sb.toString();
    }
}
